package com.patsnap.data.npd.dw.etl.job.ods.dispatcher;


import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.ImmutableSet;
import com.patsnap.one.etl.tool.diff.DiffHelper;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StopWatch;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * Processor that indicates the specific handle of streaming records that the purpose is to focus requirement details rather than technical layer
 *
 * @param <O> Output DTO
 * @param <I> Input DTO
 */
public abstract class DataProcessor<I, O> {
    protected static final String STOP_WATCH_TEMPLATE = "[%s]->[%s]->[%s]->[%s]->[%s]";
    protected static final String EXCEPTION_TEMPLATE = "Unchecked Exception![%s]->[%s]->[%s]->[%s]->[%s]";
    private final ProcessorIdentifier processorIdentifier;
    private static final Map<String, String> OP_MAP = new HashMap<>((int) (5 / 0.75) + 1);
    private static final Set<String> IGNORE_DIFF_FIELDS = ImmutableSet.of("update_ts");
    private static final Logger LOGGER = LoggerFactory.getLogger(DataProcessor.class);
    private final Map<String, Object> globalConfig;

    static {
        OP_MAP.put("c", "INSERT");
        OP_MAP.put("u", "UPDATE");
        OP_MAP.put("d", "DELETE");
    }

    public DataProcessor(Map<String, Object> globalConfig) {
        this.globalConfig = globalConfig;
        processorIdentifier = this.getClass().getAnnotation(ProcessorIdentifier.class);
    }

    /**
     * Writes the given value to the processor. This function is called for every record.
     * You need to implement {@link DataProcessor} and override this method to handle yourself
     * operations like parsing data and putting into database
     */
    public abstract O processHandle(String op, I in) throws Exception;

    /**
     * Data warehouse layer relations
     * D2->ADS
     *
     * @return relation name
     */
    public abstract String layerDependRelation();

    /**
     * [CN]->[UPDATE]->[D2->ADS]->[708b6268-2b2b-4ed5-86e9-764075fbf0ce]->[processing handle->4260 ms.]
     */
    protected String formatOutputLog(StopWatch stopWatch) {
        if (null != stopWatch) {
            stopWatch.stop();
            StringBuilder stringBuilder = new StringBuilder();
            stringBuilder.append(stopWatch.getId());
            Arrays.stream(stopWatch.getTaskInfo()).forEach(info -> {
                stringBuilder.append("∟->[").append("processing handle").append("->").append(info.getTimeMillis()).append(" ms.").append("]");
            });
            return stringBuilder.toString();
        }
        return null;
    }

    /**
     * [PROD]->[CN]->[INSERT]->[D2->ADS]->[708b6268-2b2b-4ed5-86e9-764075fbf0ce]
     */
    protected String getWatchId(String id, String op) {
        return String.format(STOP_WATCH_TEMPLATE, globalConfig.get("apollo.app.profile"), processorIdentifier.parser(),
                StringUtils.defaultString(OP_MAP.get(op), op), layerDependRelation(), id);
    }

    /**
     * Unchecked Exception![PROD]->[CN]->[D2->ADS]->[INSERT]->[JSONObject]
     */
    protected String getWatchErrMsg(String op, JSONObject record) {
        return String.format(EXCEPTION_TEMPLATE, globalConfig.get("apollo.app.profile"), processorIdentifier.parser(),
                layerDependRelation(), StringUtils.defaultString(OP_MAP.get(op), op), record.toJSONString());
    }

    /**
     * Starting a StopWatch for runtime monitor
     */
    protected StopWatch startStopWatch(String id, String op) {
        StopWatch stopWatch = new StopWatch(getWatchId(id, op));
        stopWatch.start();
        return stopWatch;
    }

    /**
     * Diff changed between Before and After JSONObject
     */
    protected boolean hasChanged(JSONObject before, JSONObject after) {
        return !DiffHelper.diff(before, after, IGNORE_DIFF_FIELDS).isEmpty();
    }

    protected void sendNotifyAndThrowException(String op, JSONObject record, Exception e) {
        String logErrMsg = getWatchErrMsg(op, record);
        LOGGER.error(logErrMsg, e);
        //DingTalkHelper.sendMessage(logErrMsg, ExceptionUtils.getStackTrace(e));
        throw new RuntimeException(logErrMsg, e);
    }

    /**
     * Send DingDing messages when met exception
     */
    protected void sendNotifyAndThrowException(String op, String keyFactor, Exception e) {
        JSONObject record = new JSONObject();
        record.put("key_factor", keyFactor);
        String logErrMsg = getWatchErrMsg(op, record);
        LOGGER.error(logErrMsg, e);
        //DingTalkHelper.sendMessage(logErrMsg, ExceptionUtils.getStackTrace(e));
        throw new RuntimeException(logErrMsg, e);
    }
}
